package com.we.mexico.flink.loan.dim;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.mexico.flink.entity.*;
import com.we.mexico.flink.function.FilterContainsKeyFunction;
import com.we.mexico.flink.function.MapToEntityFunction;
import com.we.mexico.flink.utils.FlinkDataStreamUtils;
import com.we.mexico.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.*;

/**
 * DIM 层 墨西哥风控用户借款还款需求;
 *
 * @author jasonlee
 */
public class DimBorrowInfoLabel {
    public static Logger LOG = LoggerFactory.getLogger(DimBorrowInfoLabel.class);
    public static final String RELEASEPROP = "loanlabel/dim_loan_label.properties";
    public static final String KEY_BY = "user_key";

    /** 同时消费这几个 topic 的数据 */
    private static final List<String> TOPIC_LIST =
            Arrays.asList(
                    "fdm_hhl01_user_pid",
                    "fdm_hhl01_user_loan",
                    "fdm_hhl01_user_loan_lend",
                    "fdm_hhl01_user_audit",
                    "fdm_uni01_repay_plan");

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        WeKafkaPropertyReader propReader = WeKafkaPropertyReader.init(RELEASEPROP);
        String kfkBootStrapServer = propReader.getTmpKfkBootStrapServer();
        String kfkTopic = propReader.getTmpKfkTopic();
        int kfkPartitions = propReader.getTmpKfkPartitions();

        // sink kafka property
        Properties sinkProp = new Properties();
        sinkProp.setProperty("bootstrap.servers", kfkBootStrapServer);

        Properties consumProp = new Properties();
        consumProp.setProperty("group.id", propReader.getTmpKfkGroupId());
        consumProp.setProperty(
                "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumProp.setProperty(
                "value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaSource<String> kfkSource =
                KafkaSource.<String>builder()
                        .setBootstrapServers(propReader.getKfkBootStrapServer())
                        .setTopics(TOPIC_LIST)
                        .setStartingOffsets(OffsetsInitializer.earliest())
                        .setProperties(consumProp)
                        .setValueOnlyDeserializer(new SimpleStringSchema())
                        .build();

        SingleOutputStreamOperator<String> dataStream =
                env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "dim_user_loan")
                        .uid("dim_user_loan");

        // sink to Kafka
        FlinkKafkaProducer<String> userlabelKafkaProducer =
                new FlinkKafkaProducer<String>(
                        kfkTopic,
                        new WeKafkaKeyedSerializationSchema(),
                        sinkProp,
                        java.util.Optional.of(new WeKafkaCustomPartitioner()));

        // 根据借款时间取最新一条数据
        final DataStream<UserLoanInfo> userLoanDataStream =
                dataStream
                        .filter(new FilterContainsKeyFunction<>("borrow_money"))
                        .map(new MapToEntityFunction<String, UserLoanInfo>(UserLoanInfo.class))
                        .keyBy(data -> data.getUser_key())
                        .process(
                                new KeyedProcessFunction<String, UserLoanInfo, UserLoanInfo>() {
                                    transient ValueState<UserLoanInfo> borrowInfoValueState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        borrowInfoValueState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<
                                                                        UserLoanInfo>(
                                                                        "BorrowInfo",
                                                                        UserLoanInfo.class));
                                    }

                                    @Override
                                    public void processElement(
                                            UserLoanInfo value,
                                            KeyedProcessFunction<String, UserLoanInfo, UserLoanInfo>
                                                            .Context
                                                    ctx,
                                            Collector<UserLoanInfo> out) {
                                        try {

                                            final UserLoanInfo lastBorrowInfo =
                                                    borrowInfoValueState.value();
                                            // 第一条数据进来先更新状态
                                            if (lastBorrowInfo == null) {
                                                borrowInfoValueState.update(value);
                                                out.collect(value);
                                            } else {
                                                // 最新的一条
                                                if (value.getBorrow_time()
                                                        > lastBorrowInfo.getBorrow_time()) {
                                                    borrowInfoValueState.update(value);
                                                    out.collect(value);
                                                }
                                            }
                                        } catch (Exception e) {
                                            LOG.error(
                                                    "UserLoanInfo 方法失败了 !" + e + "原始数据为: " + value);
                                        }
                                    }
                                });

        // 根据借款时间取最新一条数据
        final SingleOutputStreamOperator<UserLoanLendInfo> userLoanLendDataStream =
                dataStream
                        .filter(new FilterContainsKeyFunction<>("loan_purpose"))
                        .map(
                                new MapToEntityFunction<String, UserLoanLendInfo>(
                                        UserLoanLendInfo.class))
                        .keyBy(data -> data.getUser_key())
                        .process(
                                new KeyedProcessFunction<
                                        String, UserLoanLendInfo, UserLoanLendInfo>() {
                                    transient ValueState<UserLoanLendInfo> borrowInfoValueState1;

                                    @Override
                                    public void open(Configuration parameters) {
                                        borrowInfoValueState1 =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<
                                                                        UserLoanLendInfo>(
                                                                        "UserLoanLendInfo",
                                                                        UserLoanLendInfo.class));
                                    }

                                    @Override
                                    public void processElement(
                                            UserLoanLendInfo value,
                                            KeyedProcessFunction<
                                                                    String,
                                                                    UserLoanLendInfo,
                                                                    UserLoanLendInfo>
                                                            .Context
                                                    ctx,
                                            Collector<UserLoanLendInfo> out) {
                                        try {

                                            final UserLoanLendInfo lastBorrowInfo =
                                                    borrowInfoValueState1.value();
                                            // 第一条数据进来先更新状态
                                            if (lastBorrowInfo == null) {
                                                borrowInfoValueState1.update(value);
                                                out.collect(value);
                                            } else {
                                                // 最新的一条
                                                if (value.getCreated_at()
                                                        > lastBorrowInfo.getCreated_at()) {
                                                    borrowInfoValueState1.update(value);
                                                    out.collect(value);
                                                }
                                            }
                                        } catch (Exception e) {
                                            LOG.error(
                                                    "UserLoanLendInfo 解析失败了!"
                                                            + e
                                                            + "原始数据为: "
                                                            + value);
                                        }
                                    }
                                });

        final SingleOutputStreamOperator<RepayPlan> process =
                dataStream
                        .filter(new FilterContainsKeyFunction<>("payoff_time"))
                        .map(new MapToEntityFunction<String, RepayPlan>(RepayPlan.class))
                        .keyBy(data -> data.getUser_key() + data.getLoan_key())
                        .process(
                                new KeyedProcessFunction<String, RepayPlan, RepayPlan>() {
                                    // 存储未还期次,对应的应还日期
                                    transient MapState<Long, Timestamp> mapState;
                                    // 存储最大的逾期天数
                                    transient ValueState<Long> valueState;
                                    // 存储最大 payoff_time
                                    transient ValueState<Timestamp> payoffTimeState;
                                    final Map<Timestamp, Long> treeMap = new TreeMap<>();

                                    @Override
                                    public void open(Configuration parameters) {
                                        mapState =
                                                getRuntimeContext()
                                                        .getMapState(
                                                                new MapStateDescriptor<
                                                                        Long, Timestamp>(
                                                                        "mapState",
                                                                        Long.class,
                                                                        Timestamp.class));

                                        valueState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<Long>(
                                                                        "valueState", Long.class));

                                        payoffTimeState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<Timestamp>(
                                                                        "payoffTimeState",
                                                                        Timestamp.class));
                                    }

                                    @Override
                                    public void processElement(
                                            RepayPlan value,
                                            KeyedProcessFunction<String, RepayPlan, RepayPlan>
                                                            .Context
                                                    ctx,
                                            Collector<RepayPlan> out) {
                                        try {
                                            final Long lastOverDays =
                                                    valueState.value() == null
                                                            ? 0L
                                                            : valueState.value();
                                            final Timestamp lastPayoffTime =
                                                    payoffTimeState.value();
                                            // 已经结清, 结清时间 - 应还时间
                                            if (value.getPayoff_time() != null) {
                                                // 当前最大逾期天数
                                                final long maxDays =
                                                        getDays(
                                                                value.getPayoff_time(),
                                                                value.getDue_date());
                                                // 当前和状态里取最大值
                                                final long maxOverDays =
                                                        Math.max(lastOverDays, maxDays);
                                                // 把最大逾期天数更新到状态里面
                                                valueState.update(maxOverDays);
                                                value.setMax_overdays(maxOverDays);
                                                // 更新 max payoff time
                                                if (lastPayoffTime != null
                                                        && value.getPayoff_time().getTime()
                                                                > lastPayoffTime.getTime()) {
                                                    payoffTimeState.update(value.getPayoff_time());
                                                } else {
                                                    payoffTimeState.update(value.getPayoff_time());
                                                }
                                                // 已经结清,删除状态里面已结清的期次
                                                if (mapState.contains(value.getPeriod())) {
                                                    mapState.remove(value.getPeriod());
                                                }
                                            } else {
                                                // 还未结清,添加到状态里
                                                mapState.put(
                                                        value.getPeriod(), value.getDue_date());
                                            }
                                            for (Map.Entry<Long, Timestamp> entries :
                                                    mapState.entries()) {
                                                treeMap.put(entries.getValue(), entries.getKey());
                                            }
                                            // 全部结清,取最大 payoff time
                                            if (treeMap.isEmpty()) {
                                                value.setPayoff_time(payoffTimeState.value());
                                            } else {
                                                value.setPayoff_time(null);
                                            }
                                            List<Map.Entry<Timestamp, Long>> list =
                                                    new ArrayList<>(treeMap.entrySet());
                                            // 注意 list 有可能是空的,当 list 为空时,说明借款全部还清
                                            if (!list.isEmpty()) {
                                                value.setMin_over_date(list.get(0).getKey());
                                            } else {
                                                value.setMin_over_date(null);
                                            }
                                            if (value.getUser_key() != null) {
                                                out.collect(value);
                                            }
                                            // 清除数据
                                            treeMap.clear();
                                        } catch (Exception e) {
                                            LOG.error("RepayPlan 失败!" + e + "原始数据为:" + value);
                                        }
                                    }
                                });

        final SingleOutputStreamOperator<UserAuditInfo> userAuditCountDataStream =
                dataStream
                        .filter(new FilterContainsKeyFunction<>("audit_id"))
                        .map(new MapToEntityFunction<String, UserAuditInfo>(UserAuditInfo.class))
                        .keyBy(data -> data.getUser_key())
                        .process(
                                new KeyedProcessFunction<String, UserAuditInfo, UserAuditInfo>() {
                                    transient ListState<Long> listState;
                                    private final Set<Long> set = new HashSet<>();

                                    @Override
                                    public void open(Configuration parameters) {
                                        listState =
                                                getRuntimeContext()
                                                        .getListState(
                                                                new ListStateDescriptor<Long>(
                                                                        "listState", Long.class));
                                    }

                                    @Override
                                    public void processElement(
                                            UserAuditInfo value,
                                            KeyedProcessFunction<
                                                                    String,
                                                                    UserAuditInfo,
                                                                    UserAuditInfo>
                                                            .Context
                                                    ctx,
                                            Collector<UserAuditInfo> out) {
                                        try {
                                            listState.add(value.getAudit_id());
                                            listState
                                                    .get()
                                                    .forEach(
                                                            data -> {
                                                                set.add(data);
                                                            });
                                            value.setAudit_count(set.size());
                                            out.collect(value);
                                            set.clear();
                                        } catch (Exception e) {
                                            LOG.error(
                                                    "UserAuditInfo 解析失败了 " + e + "原始数据为: " + value);
                                        }
                                    }
                                });

        // user_loan
        FlinkDataStreamUtils.mapToJSONString(userLoanDataStream, "borrow")
                .addSink(userlabelKafkaProducer)
                .setParallelism(kfkPartitions);

        // user_loan_lend
        FlinkDataStreamUtils.mapToJSONString(userLoanLendDataStream, "borrow")
                .addSink(userlabelKafkaProducer)
                .setParallelism(kfkPartitions);

        // 进件次数写到 kafka
        FlinkDataStreamUtils.mapToJSONString(userAuditCountDataStream, "borrow")
                .addSink(userlabelKafkaProducer)
                .setParallelism(kfkPartitions);

        // id_number 写到 kafka
        FlinkDataStreamUtils.mapToJSONString(
                        dataStream
                                .filter(new FilterContainsKeyFunction<>("id_number"))
                                .map(
                                        new MapToEntityFunction<String, UserIdNumberInfo>(
                                                UserIdNumberInfo.class)),
                        "common")
                .addSink(userlabelKafkaProducer)
                .setParallelism(kfkPartitions);

        // 还款计划写到 kafka
        FlinkDataStreamUtils.mapToJSONString(process, "repay")
                .addSink(userlabelKafkaProducer)
                .setParallelism(kfkPartitions);

        env.execute(DimBorrowInfoLabel.class.getSimpleName());
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            return jsonObject.getString(KEY_BY).getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            return Math.abs(new String(key).hashCode() % partitions.length);
        }
    }

    private static long getDays(Timestamp time1, Timestamp time2) {
        return (time1.getTime() - time2.getTime()) / 86400000;
    }

    @Deprecated
    private static long timeToLong(String time) {
        return Timestamp.valueOf(time).getTime();
    }
}
